﻿using MQTTnet.Packets;
using MQTTnet.Protocol;
using MQTTnet.Server;
using System.Net;
using System.Text;
using ZX.MQTTServer.Common;

namespace ZX.MQTTServer
{
    internal class Program
    {
        private static MqttServer mqttServer = null;
        private static SubscriptionPersistence _persistence = null;

        static async Task Main(string[] args)
        {
            await ServerStart();

            Console.WriteLine("MQTT服务已启动!");
            Console.ReadKey();
        }

        private static async Task ServerStart()
        {
            _persistence = new SubscriptionPersistence();

            MqttServerOptions mqttServerOptions = new MqttServerOptionsBuilder()
                .WithDefaultEndpoint()  // 必须启用默认终端节点                 
                .WithDefaultEndpointBoundIPAddress(IPAddress.Any) // 和0.0.0.0一样://.WithDefaultEndpointBoundIPAddress(IPAddress.Parse("0.0.0.0"))
                .WithDefaultEndpointPort(1883)
                .WithPersistentSessions() // 持久化会话
                .Build();


            mqttServer = new MqttServerFactory().CreateMqttServer(mqttServerOptions);


            mqttServer.StartedAsync += e =>
            {
                var allSubs = _persistence.GetAll();
                Console.WriteLine($"恢复{allSubs.Count}个订阅...");

                foreach (var sub in allSubs)
                {
                    // 手动为服务器"订阅"主题(用于转发)
                    mqttServer.SubscribeAsync(sub.ClientId, new[]
                    {
                        new MqttTopicFilter
                        {
                            Topic=sub.Topic,
                            QualityOfServiceLevel=sub.Qos
                        }
                    });

                    Console.WriteLine($"恢复客户端id:{sub.ClientId},的主题:{sub.Topic},Qos:{sub.Qos}");
                }

                return Task.CompletedTask;
            };

            // 客户端连接时,恢复其订阅
            mqttServer.ClientConnectedAsync += e =>
            {
                Console.WriteLine($">>> 客户端:{e.ClientId},连接");

                var clientId = e.ClientId;
                var subs = _persistence.GetByClient(clientId);

                foreach (var sub in subs)
                {
                    mqttServer.SubscribeAsync(clientId, new[]
                    {
                        new MqttTopicFilter
                        {
                            Topic=sub.Topic,
                            QualityOfServiceLevel=sub.Qos
                        }
                    });

                    Console.WriteLine($"恢复客户端id:{sub.ClientId},的主题:{sub.Topic},Qos:{sub.Qos}");
                }
                Console.WriteLine($"{clientId}连接,恢复{subs.Count}个订阅");

                return Task.CompletedTask;
            };

            // 客户端断开连接时
            mqttServer.ClientDisconnectedAsync += e =>
            {
                Console.WriteLine($">>> 客户端:{e.ClientId},断开连接,原因:{e.DisconnectType.ToString()}");
                return Task.CompletedTask;
            };


            // 客户端成功订阅时:持久化
            mqttServer.ClientSubscribedTopicAsync += e =>
            {
                var clientId = e.ClientId;
                var topic = e.TopicFilter.Topic;
                var qos = e.TopicFilter.QualityOfServiceLevel;

                _persistence.AddOrUpdate(clientId, topic, qos);

                Console.WriteLine($"持久化订阅: {clientId} -> {topic} (Qos{qos})");
                return Task.CompletedTask;
            };

            // 取消订阅时:清除
            mqttServer.ClientUnsubscribedTopicAsync += e =>
            {
                _persistence.Remove(e.ClientId, e.TopicFilter);
                Console.WriteLine($"清除订阅:{e.ClientId}->{e.TopicFilter}");
                return Task.CompletedTask;
            };

            // 客户端连接验证事件
            mqttServer.ValidatingConnectionAsync += e =>
            {
                if (e.UserName != "admin" || e.Password != "123456")
                {
                    e.ReasonCode = MqttConnectReasonCode.BadUserNameOrPassword;
                    return Task.CompletedTask;
                }

                e.ReasonCode = MqttConnectReasonCode.Success;

                return Task.CompletedTask;
            };


            //关键：拦截客户端发布的消息（相当于旧版的 ApplicationMessageReceivedAsync）
            mqttServer.InterceptingPublishAsync += context =>
            {
                // 第一步：先允许发布，避免被丢弃
                // 必须设置 ProcessPublish = true 否则消息不会被处理（也不会转发给订阅者）,false表示丢弃消息
                context.ProcessPublish = true; // 允许发布

                // 获取消息内容
                var clientId = context.ClientId;
                var topic = context.ApplicationMessage.Topic;
                var payload = context.ApplicationMessage.Payload;
                var qos = context.ApplicationMessage.QualityOfServiceLevel;
                bool retain = context.ApplicationMessage.Retain;

                string message = payload.Length == 0 ? "" : Encoding.UTF8.GetString(payload);

                Console.WriteLine($">>> 收到消息 - 客户端:{clientId}, 主题:{topic}, 消息:{message}, QoS:{qos}, Retain:{retain}");


                return Task.CompletedTask;
            };



            // 监控消息入队或丢弃情况
            mqttServer.ApplicationMessageEnqueuedOrDroppedAsync += e =>
            {
                Console.WriteLine($">>> 发送者:{e.SenderClientId} -> 接收者:{e.ReceiverClientId}");

                // 转换负载这个字节数组为字符串
                //Console.WriteLine(e.ApplicationMessage.ConvertPayloadToString());

                // 接收字节数组
                //foreach (byte item in e.ApplicationMessage.Payload.ToArray())
                //{
                //    Console.Write(item.ToString("X2")+" ");
                //}

                if (e.IsDropped)
                {
                    Console.WriteLine($">>> 消息已丢弃,主题:{e.ApplicationMessage.Topic}," +
                        $"负载:{System.Text.Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}");
                }
                else
                {
                    Console.WriteLine($">>> 消息已入队,主题:{e.ApplicationMessage.Topic}," +
                        $"负载:{System.Text.Encoding.UTF8.GetString(e.ApplicationMessage.Payload)}");
                }


                return Task.CompletedTask;
            };

            await mqttServer.StartAsync();
        }
    }
}
